package net.wicp.tams.commons.zookeeper;

import java.util.List;
import java.util.Properties;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSONObject;

import lombok.extern.slf4j.Slf4j;
import net.wicp.tams.commons.Conf;
import net.wicp.tams.commons.Conf.Callback;
import net.wicp.tams.commons.apiext.SVNUtil;
import net.wicp.tams.commons.apiext.StringUtil;

@Slf4j
public class ZK {
	private static volatile ZK INSTANCE;
	private final ZooKeeper zk;

	static {
		Conf.addCallBack("zookeeper", new Callback() {
			@Override
			public void doReshConf(Properties newProperties) {
				destroy();
			}
		}, "zookeeper.server%s");
	}

	private ZK() {
		try {
			String canReadOnlyStr = Conf.get("zookeeper.server.canbereadonly");
			boolean canReadOnly = StringUtil.isNull(canReadOnlyStr) ? false : Boolean.valueOf(canReadOnlyStr);
			String watcherStr = Conf.get("zookeeper.server.watcher");
			Watcher watcher = StringUtil.isNull(watcherStr) ? new Watcher() {
				// 监控所有被触发的事件
				@Override
				public void process(WatchedEvent event) {
					// System.out.println("已经触发了" + event.getType() + "事件！");
				}
			} : (Watcher) Class.forName(watcherStr).newInstance();

			zk = new ZooKeeper(Conf.get("zookeeper.server.constr"),
					Integer.parseInt(Conf.get("zookeeper.server.timeout")), watcher, canReadOnly);
		} catch (Exception e) {
			throw new IllegalArgumentException("初始化zk实例出错，请检查配置信息");
		}
	}

	public static void destroy() {
		if (INSTANCE != null && INSTANCE.zk != null) {
			try {
				INSTANCE.zk.close();
				INSTANCE = null;
			} catch (InterruptedException e) {
				log.error("关闭zk连接出错。");
			}
		}

	}

	public static final ZK getInstance() {
		if (INSTANCE == null || INSTANCE.zk == null || !INSTANCE.zk.getState().isAlive()
				|| !INSTANCE.zk.getState().isConnected()) {
			synchronized (ZK.class) {
				if (INSTANCE == null || INSTANCE.zk == null || !INSTANCE.zk.getState().isAlive()
						|| !INSTANCE.zk.getState().isConnected()) {
					if (INSTANCE != null && INSTANCE.zk != null) {
						try {
							INSTANCE.zk.close();
						} catch (InterruptedException e) {
							log.error("关闭ZK实例异常");
							throw new IllegalAccessError("关闭ZK实例异常");
						}
					}
					INSTANCE = new ZK();
				}
			}
		}
		return INSTANCE;
	}

	public ZooKeeper getZk() {
		return zk;
	}

	public void createMultilevelNode(String path) {
		if (StringUtil.isNull(path) || !path.startsWith("/")) {
			throw new IllegalArgumentException("path错误");
		}
		path = path.replace("\\", "/");
		String[] paths = path.split("/");

		try {
			for (String ele : paths) {
				if (StringUtil.isNull(ele)) {
					continue;
				}
				String tempstr = "/" + ele;
				int idx = path.indexOf(tempstr);
				String tempPath = path.substring(0, idx + tempstr.length());
				Stat stat = zk.exists(tempPath, false);
				if (stat == null) {
					zk.create(tempPath, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
				}
			}
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
	}

	public String createNode(String path, String value) {
		path = StringUtil.trimSpace(path);
		if (StringUtil.isNull(path)) {
			return null;
		}

		value = StringUtil.trimSpace(value);
		value = StringUtil.isNull(value) ? null : value;
		try {
			String pathret = null;
			if (value == null) {
				pathret = zk.create(path, null, Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			} else {
				pathret = zk.create(path, value.getBytes("UTF-8"), Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			}
			return pathret;
		} catch (Exception e) {
			log.error("创建结点错误", e);
			return null;
		}
	}

	public String createNode(String path) {
		return createNode(path, null);
	}

	public Stat updateNode(String path, String data) {
		try {
			return zk.setData(path, data.getBytes("UTF-8"), -1);
		} catch (Exception e) {
			log.error("更新数据错误", e);
			return null;
		}
	}

	public String getData(String path, boolean watcher) {
		try {
			byte[] by = zk.getData(path, watcher, null);
			return by == null ? null : new String(by, "UTF-8");
		} catch (Exception e) {
			log.error("获取数据错误", e);
			throw new IllegalArgumentException("获取数据错误", e);
		}
	}

	/**
	 * 得到zk节点数据，没有监听
	 * 
	 * @param path
	 *            zk路径
	 * @return
	 */
	public String getData(String path) {
		return getData(path, false);
	}

	/***
	 * 把zk节点数据转成json数据，没有监听
	 * 
	 * @param path
	 *            zk路径
	 * @return
	 */
	public JSONObject getDataJson(String path) {
		return JSONObject.parseObject(getData(path));
	}

	/***
	 * 把zk节点数据转成java对象，没有监听
	 * 
	 * @param path
	 *            zk路径
	 * @param clazz
	 *            要转成对象的class
	 * @return
	 */
	public <T> T getDateObj(String path, Class<T> clazz) {
		return JSONObject.parseObject(getData(path), clazz);
	}

	public void deleteNode(String path) {
		try {
			zk.delete(path, -1);
		} catch (Exception e) {
			log.error("删除数据错误", e);
			throw new IllegalArgumentException("删除数据错误", e);
		}
	}

	/****
	 * 找到一个目录下最小的节点，适用于 PERSISTENT_SEQUENTIAL和 EPHEMERAL_SEQUENTIAL 节点
	 * 
	 * @param dir
	 *            要查找的目录
	 * @param prefix
	 *            前缀
	 * @return 最小的节点名
	 * @throws KeeperException
	 * @throws InterruptedException
	 */
	public String smallestChildName(String dir, String prefix) throws KeeperException, InterruptedException {
		long minId = Long.MAX_VALUE;
		String minName = "";
		prefix = StringUtil.hasNull(prefix, null);
		List<String> childNames = null;

		try {
			childNames = zk.getChildren(dir, false);
		} catch (KeeperException.NoNodeException e) {
			log.warn("Caught: " + e, e);
			return null;
		}

		for (String childName : childNames) {
			try {
				// Check format
				if (prefix != null && !childName.regionMatches(0, prefix, 0, prefix.length())) {
					log.warn("Found child node with improper name: " + childName);
					continue;
				}
				String suffix = childName.substring(prefix == null ? 0 : prefix.length());
				long childId = Long.parseLong(suffix);
				if (childId < minId) {
					minId = childId;
					minName = childName;
				}
			} catch (NumberFormatException e) {
				log.warn("Found child node with improper format : " + childName + " " + e, e);
			}
		}

		if (minId < Long.MAX_VALUE) {
			return minName;
		} else {
			return null;
		}
	}

}
